/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *  http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.gravitino.storage.relational.service;

import static org.apache.gravitino.metrics.source.MetricsSource.GRAVITINO_RELATIONAL_STORE_METRIC_NAME;

import java.io.IOException;
import java.util.List;
import java.util.Locale;
import java.util.stream.Collectors;
import org.apache.gravitino.Entity;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.exceptions.IllegalNamespaceException;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.job.JobHandle;
import org.apache.gravitino.meta.JobEntity;
import org.apache.gravitino.metrics.Monitored;
import org.apache.gravitino.storage.relational.mapper.JobMetaMapper;
import org.apache.gravitino.storage.relational.po.JobPO;
import org.apache.gravitino.storage.relational.utils.ExceptionUtils;
import org.apache.gravitino.storage.relational.utils.SessionUtils;
import org.apache.gravitino.utils.NamespaceUtil;

public class JobMetaService {

  private static final JobMetaService INSTANCE = new JobMetaService();

  private JobMetaService() {
    // Private constructor to prevent instantiation
  }

  public static JobMetaService getInstance() {
    return INSTANCE;
  }

  @Monitored(
      metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
      baseMetricName = "listJobsByNamespace")
  public List<JobEntity> listJobsByNamespace(Namespace ns) {
    String metalakeName = ns.level(0);
    if (ns.length() == 3) {
      // If the namespace is job namespace, we will list all the jobs in the metalake.
      List<JobPO> jobPOs =
          SessionUtils.getWithoutCommit(
              JobMetaMapper.class, mapper -> mapper.listJobPOsByMetalake(metalakeName));
      return jobPOs.stream().map(po -> JobPO.fromJobPO(po, ns)).collect(Collectors.toList());

    } else if (ns.length() == 4) {
      // If the namespace is generated by a job template identifier, we will list all the jobs
      // associate with the job template.
      NameIdentifier jobTemplateIdent = NameIdentifier.of(ns.levels());
      String jobTemplateName = jobTemplateIdent.name();
      List<JobPO> jobPOs =
          SessionUtils.getWithoutCommit(
              JobMetaMapper.class,
              mapper -> mapper.listJobPOsByMetalakeAndTemplate(metalakeName, jobTemplateName));
      return jobPOs.stream()
          .map(po -> JobPO.fromJobPO(po, NamespaceUtil.ofJob(metalakeName)))
          .collect(Collectors.toList());

    } else {
      throw new IllegalNamespaceException("Invalid namespace for listing jobs: %s", ns);
    }
  }

  @Monitored(
      metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
      baseMetricName = "getJobByIdentifier")
  public JobEntity getJobByIdentifier(NameIdentifier ident) {
    String metalakeName = ident.namespace().level(0);
    String jobRunId = ident.name();
    long jobRunIdLong;
    try {
      jobRunIdLong = Long.parseLong(jobRunId.substring(JobHandle.JOB_ID_PREFIX.length()));
    } catch (NumberFormatException e) {
      throw new NoSuchEntityException("Invalid job run ID format %s", jobRunId);
    }

    JobPO jobPO =
        SessionUtils.getWithoutCommit(
            JobMetaMapper.class,
            mapper -> mapper.selectJobPOByMetalakeAndRunId(metalakeName, jobRunIdLong));
    if (jobPO == null) {
      throw new NoSuchEntityException(
          NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
          Entity.EntityType.JOB.name().toLowerCase(Locale.ROOT),
          jobRunId);
    }
    return JobPO.fromJobPO(jobPO, ident.namespace());
  }

  @Monitored(metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, baseMetricName = "insertJob")
  public void insertJob(JobEntity jobEntity, boolean overwrite) throws IOException {
    String metalakeName = jobEntity.namespace().level(0);

    try {
      Long metalakeId = MetalakeMetaService.getInstance().getMetalakeIdByName(metalakeName);

      JobPO.JobPOBuilder builder = JobPO.builder().withMetalakeId(metalakeId);
      JobPO jobPO = JobPO.initializeJobPO(jobEntity, builder);

      SessionUtils.doWithCommit(
          JobMetaMapper.class,
          mapper -> {
            if (overwrite) {
              mapper.insertJobMetaOnDuplicateKeyUpdate(jobPO);
            } else {
              mapper.insertJobMeta(jobPO);
            }
          });
    } catch (RuntimeException e) {
      ExceptionUtils.checkSQLException(e, Entity.EntityType.JOB, jobEntity.id().toString());
    }
  }

  @Monitored(metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, baseMetricName = "deleteJob")
  public boolean deleteJob(NameIdentifier jobIdent) {
    String jobRunId = jobIdent.name();
    long jobRunIdLong;
    try {
      jobRunIdLong = Long.parseLong(jobRunId.substring(JobHandle.JOB_ID_PREFIX.length()));
    } catch (NumberFormatException e) {
      throw new NoSuchEntityException("Invalid job run ID format %s", jobRunId);
    }
    int result =
        SessionUtils.doWithCommitAndFetchResult(
            JobMetaMapper.class, mapper -> mapper.softDeleteJobMetaByRunId(jobRunIdLong));
    return result > 0;
  }

  @Monitored(
      metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
      baseMetricName = "deleteJobsByLegacyTimeline")
  public int deleteJobsByLegacyTimeline(long legacyTimeline, int limit) {
    // Mark jobs as deleted for finished jobs, so that they can be cleaned up later
    SessionUtils.doWithCommit(
        JobMetaMapper.class, mapper -> mapper.softDeleteJobMetasByLegacyTimeline(legacyTimeline));

    return SessionUtils.doWithCommitAndFetchResult(
        JobMetaMapper.class,
        mapper -> mapper.deleteJobMetasByLegacyTimeline(legacyTimeline, limit));
  }
}
